package main

import (
	"context"
	"fmt"
	"github.com/apache/pulsar-client-go/pulsar"
	"log"
	"pulsar-go/utils"
)

var client pulsar.Client
var consumer pulsar.Consumer
var err error
var done = make(chan bool)

func main() {
	//监听
	listen()
	//驻留
	utils.Reserve()
}

func listen() {
	// 创建Pulsar客户端
	if client, err = utils.BuildClint(); err != nil {
		log.Printf("获取连接失败:%s\n", err.Error())
		return
	}
	defer client.Close()

	// 创建消费者
	if consumer, err = utils.BuildConsumer(client); err != nil {
		log.Printf("消息接收端获取连接失败:%v\n", err.Error())
		return
	}
	defer consumer.Close()
	// 消费消息
	log.Println("开始监听消息")
	go func() {
		defer func() {
			if e := recover(); e != nil {
				done <- true
				fmt.Println("发生异常:", e)
			}
		}()
		for {
			var msg pulsar.Message
			if msg, err = consumer.Receive(context.Background()); err != nil {
				done <- true
				log.Printf("接收消息失败:%v\n", err.Error())
			}
			fmt.Printf("接收到消息：%s\n", string(msg.Payload()))
			// 确认消息已处理
			consumer.Ack(msg)
		}
	}()
	<-done
}
